package frame.m.com.mylibrary.http;


import com.blankj.utilcode.util.LogUtils;
import com.google.gson.Gson;
import com.orhanobut.logger.Logger;
import com.trello.rxlifecycle2.android.ActivityEvent;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;

/**
 * Created by Administrator on 2017/7/7.
 */

public class RxHelper {
    public static <T> ObservableTransformer<BaseResult<T>, T> handleResult(final ActivityEvent activityEvent, final PublishSubject<ActivityEvent> subject) {
        return new ObservableTransformer<BaseResult<T>, T>() {
            @Override
            public ObservableSource<T> apply(@NonNull Observable<BaseResult<T>> tObservable) {
                Observable<ActivityEvent> observable=
                        subject.filter(new Predicate<ActivityEvent>() {
                            @Override
                            public boolean test(@NonNull ActivityEvent event) throws
                                    Exception {
                                return event.equals(activityEvent);
                            }
                        }).take(1);
                return tObservable.flatMap(new Function<BaseResult<T>, Observable<T>>() {
                    @Override
                    public Observable<T> apply(@NonNull BaseResult<T> tBaseResult) throws
                            Exception {
                        Logger.d(new Gson().toJson(tBaseResult)+">>>>"+tBaseResult.toString());
                        if (tBaseResult.errorCode .equals(HttpConstant.SUCCESS) ) {
                                return create(tBaseResult.data);
                        } else {
                            return Observable.error(new ApiException(tBaseResult.errorMsg));
                        }
                    }
                }).takeUntil(observable).subscribeOn(Schedulers.io()).unsubscribeOn(Schedulers.io()).subscribeOn
                        (AndroidSchedulers.mainThread()).observeOn(AndroidSchedulers.mainThread());
            }
        };
    }

    public static <T> Observable<T> create(final T data) {
        return Observable.create(new ObservableOnSubscribe<T>() {
            @Override
            public void subscribe(ObservableEmitter<T> subscriber) throws Exception {
                try {
                    Object o;
                    if (data==null){
                        o = "";
                    }else
                    {
                        o=data;
                    }
                    subscriber.onNext(((T) o));
                    subscriber.onComplete();
                } catch (Exception e) {
                    subscriber.onError(e);
                }

            }
        });
    }
}
